#include <Processors/Formats/Impl/ProtobufRowOutputFormat.h>

#if USE_PROTOBUF
#   include <Formats/FormatFactory.h>
#   include <Core/Block.h>
#   include <Formats/FormatSettings.h>
#   include <Formats/ProtobufSchemas.h>
#   include <Formats/ProtobufSerializer.h>
#   include <Formats/ProtobufWriter.h>
#   include <Processors/Port.h>

#   include <google/protobuf/descriptor.h>

namespace DB
{
namespace ErrorCodes
{
    extern const int NO_ROW_DELIMITER;
}

ProtobufRowOutputFormat::ProtobufRowOutputFormat(
    WriteBuffer & out_,
    SharedHeader header_,
    const ProtobufSchemaInfo & schema_info_,
    const FormatSettings & settings_,
    bool with_length_delimiter_)
    : IRowOutputFormat(header_, out_)
    , writer(std::make_unique<ProtobufWriter>(out))
    , descriptor_holder(ProtobufSchemas::instance().getMessageTypeForFormatSchema(
          schema_info_.getSchemaInfo(), ProtobufSchemas::WithEnvelope::No, settings_.protobuf.google_protos_path))
    , serializer(ProtobufSerializer::create(
          header_->getNames(),
          header_->getDataTypes(),
          descriptor_holder,
          with_length_delimiter_,
          /* with_envelope = */ false,
          settings_.protobuf.output_nullables_with_google_wrappers,
          *writer))
    , allow_multiple_rows(with_length_delimiter_ || settings_.protobuf.allow_multiple_rows_without_delimiter)
{
}

void ProtobufRowOutputFormat::write(const Columns & columns, size_t row_num)
{
    if (!allow_multiple_rows && !first_row)
        throw Exception(ErrorCodes::NO_ROW_DELIMITER,
                        "The ProtobufSingle format can't be used "
                        "to write multiple rows because this format doesn't have any row delimiter.");

    if (row_num == 0)
        serializer->setColumns(columns.data(), columns.size());

    serializer->writeRow(row_num);
}

void registerOutputFormatProtobuf(FormatFactory & factory)
{
    for (bool with_length_delimiter : {false, true})
    {
        const char * name = with_length_delimiter ? "Protobuf" : "ProtobufSingle";
        factory.registerOutputFormat(
            name,
            [with_length_delimiter](WriteBuffer & buf,
               const Block & header,
               const FormatSettings & settings,
               FormatFilterInfoPtr /*format_filter_info*/)
            {
                return std::make_shared<ProtobufRowOutputFormat>(
                    buf, std::make_shared<const Block>(header), ProtobufSchemaInfo(settings, "Protobuf", header, settings.protobuf.use_autogenerated_schema),
                    settings, with_length_delimiter);
            });
        factory.markOutputFormatNotTTYFriendly(name);
        factory.setContentType(name, "application/octet-stream");
    }
}

}

#else

namespace DB
{
    class FormatFactory;
    void registerOutputFormatProtobuf(FormatFactory &) {}
}

#endif
